/*
 * To change this template, choose Tools | Templates
 * and open the template in the editor.
 */

/*
 * FirstMapper.java
 *
 * Created on Jul 9, 2010, 4:13:51 PM
 */
package homomorphism.list;

 
import homomorphism.HomomorphismBase;
import java.lang.reflect.*;

import utility.LongPair;
import java.io.IOException;
import java.util.ArrayList;
import java.util.logging.Level;
import java.util.logging.Logger;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.mapred.MapReduceBase;
import org.apache.hadoop.mapred.Mapper;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.Reporter;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.mapred.JobConf;
 

/**
 *
 * @author takeshi
 */
//abstract
public class FirstMapper< InVal, OtVal> extends MapReduceBase implements Mapper<LongWritable, InVal, LongPair, OtVal> {

    public static final Log LOG = LogFactory.getLog(FirstMapper.class);
    private static final String userClassPro = HomomorphismBase.UserClassNameProperty;
  //  private static final String InValClassPro = HomomorphismBase.IndexKeyClassProperty;//"mapred.input.value.class";
  //  private static final String OtValClassPro = HomomorphismBase.IndexKeyClassProperty;//"mapred.output.value.class";
    public static final int Mega = HomomorphismBase.Mega;

  //  private Class FilterType=null;
    private Class AggType =null;
    private String UserClassName=null;
    private long pid = -1;
    private ArrayList<InVal> localCache;

        @Override
    public void configure(JobConf job) {

        super.configure(job);
        UserClassName = job.get(userClassPro);
//        try {
//            if (job.get(InValClassPro) != null && job.get(InValClassPro).length() > 0) {
//                FilterType = Class.forName(job.get(InValClassPro));
//            }else{
//                try {
//                    throw new IOException("1null  configure:userClassPro\n");
//                } catch (IOException ex) {
//                    Logger.getLogger(FirstMapper.class.getName()).log(Level.SEVERE, null, ex);
//                }
//            }
//
//        } catch (ClassNotFoundException ex) {
//            Logger.getLogger(FirstMapper.class.getName()).log(Level.SEVERE, null, ex);
//        }
        AggType = job.getMapOutputValueClass(); //==IndexKeyClassProperty  assertThat()...

    }

    public OtVal filter(InVal in) throws ClassNotFoundException, NoSuchMethodException, InstantiationException, IllegalAccessException, IllegalArgumentException, InvocationTargetException {
    
        OtVal rst = null;
        Class cls = Class.forName(UserClassName);
        Class partypes[] = new Class[1];
        partypes[0] =  in.getClass(); // FilterType;
        try{
            Method _filter = cls.getMethod("filter", partypes);
            if (_filter != null) {
                Object arglist[] = new Object[1];
                arglist[0] = in;
                Constructor ct = cls.getConstructor(); //partypes
                Object methobj = ct.newInstance();
                rst = (OtVal) _filter.invoke(methobj, arglist);
            }
        }catch (NoSuchMethodException e){
            //no such a method then do nothing
            rst = (OtVal)in;
        }
        return rst;
    }

    public void map(LongWritable key, InVal value, OutputCollector<LongPair, OtVal> output, Reporter reporter) throws IOException {
        long OrIndex =  key.get() ;
     //   if(pid <0)
         pid =  makePid(OrIndex) ;//(OrIndex / ThirdHomomorphismWrapper.Mega);

       // localCache.add(value);
        LongPair outKeyPair = new LongPair(pid, OrIndex);
        try {
            output.collect(outKeyPair, filter(value));
        } catch (ClassNotFoundException ex) {
            Logger.getLogger(FirstMapper.class.getName()).log(Level.SEVERE, null, ex);
        } catch (NoSuchMethodException ex) {
            Logger.getLogger(FirstMapper.class.getName()).log(Level.SEVERE, null, ex);
        } catch (InstantiationException ex) {
            Logger.getLogger(FirstMapper.class.getName()).log(Level.SEVERE, null, ex);
        } catch (IllegalAccessException ex) {
            Logger.getLogger(FirstMapper.class.getName()).log(Level.SEVERE, null, ex);
        } catch (IllegalArgumentException ex) {
            Logger.getLogger(FirstMapper.class.getName()).log(Level.SEVERE, null, ex);
        } catch (InvocationTargetException ex) {
            Logger.getLogger(FirstMapper.class.getName()).log(Level.SEVERE, null, ex);
        }
    }

    static public int makePid(long id) {
        int pid = (int) id >> 10;
        return pid;
    }
}
